
#include "../iface/debug.h"
#include <apr_network_io.h>
#include "../iface/APRInterface.h"
#include "../aprtools/APRFuture.h"
#include "TCPClient.h"

// TCPClient using blocking sockets so that the while (1) loops don't
// spin!

/** Invalidate the current session. */
void TCPClient::invalidate()
{
   errprint("TCP client state invalid");
   apr_status_t status;
   status = apr_socket_close(tsocket);
   CheckAPRError(status);
   if (clientbb) delete clientbb;
   apr_pool_clear(mp);
   apr_pool_destroy(mp);
   mp = NULL;
}

/** Constructor for the basic TCP client class. This connects
 * immediately.
 * 
 * @param ihostname Remote host name to connect to.
 * 
 * @param iportno Remote TCP port number to connect to.
 * 
 * @param imp Memory pool from which to derived the memory pool for
 * this class. */
TCPClient::TCPClient(const char *ihostname, int iportno, apr_pool_t *imp)
{
   int isinvalid = 0;
   int hassocket = 0;
   int hasmp = 0;
   clientbb = NULL;
   apr_status_t status;
   status = apr_pool_create(&mp, imp);
   CheckAPRError(status); isinvalid |= status;
   hasmp = 1;

   status = apr_sockaddr_info_get(&addrstruct, ihostname, APR_UNSPEC, iportno, 0, mp);
   CheckAPRError(status); isinvalid |= status; if (isinvalid) goto failed;
   status = apr_socket_create(&tsocket, addrstruct->family, SOCK_STREAM, APR_PROTO_TCP, mp);
   CheckAPRError(status); isinvalid |= status; if (isinvalid) goto failed;
   hassocket = 1;
   status = apr_socket_connect(tsocket, addrstruct);
   CheckAPRError(status); isinvalid |= status; if (isinvalid) goto failed;
   status = apr_socket_opt_set(tsocket, APR_SO_NONBLOCK, 0);
   CheckAPRError(status); isinvalid |= status;
   status = apr_socket_opt_set(tsocket, APR_TCP_NODELAY, 0);
   CheckAPRError(status); isinvalid |= status;
   status = apr_socket_timeout_set(tsocket, CLIENT_SOCKET_TIMEOUT * 1000);
   CheckAPRError(status); isinvalid |= status;
   if (!isinvalid)
   {
      clientbb = new APRBucketBrigade(mp);
      clientbb->addSocketBucket(tsocket);
      clientbb->writeEOS();
   }
failed:
   if (isinvalid)
   {
      errprint("TCP client failed to connect to '%s:%d'", ihostname, iportno);
      if (hassocket)
      {
         status = apr_socket_close(tsocket);
         CheckAPRError(status);
      }
      if (clientbb) delete clientbb;
      if (hasmp)
      {
         apr_pool_clear(mp);
         apr_pool_destroy(mp);
      }
      mp = NULL;
      hasmp = 0;
   }
   isvalid = !isinvalid;
}

/** Destructor for the basic TCP client class. This disconnects. */
TCPClient::~TCPClient()
{
   if (mp == NULL) return;
   apr_status_t status;
   if (isvalid)
   {
      status = apr_socket_close(tsocket);
      CheckAPRError(status);
   }
   if (clientbb) delete clientbb;
   if (mp)
   {
      apr_pool_clear(mp);
      apr_pool_destroy(mp);
      mp = NULL;
   }
}

/** Send a buffer of data over the opened connection.
 * 
 * @param buf Buffer of data.
 * 
 * @param buflen Length of the data in the buffer.
 * 
 * @return Number of bytes sent. */
apr_off_t TCPClient::sendBuffer(const char *buf, apr_off_t buflen)
{
   if (!isvalid) return 0;
   apr_status_t status;
   apr_size_t sent = 0;
   apr_off_t newlen = (buflen == -1) ? strlen(buf) : buflen;
   while (1)
   {
      apr_size_t len = newlen - sent;
      if (len == 0) break;
      status = apr_socket_send(tsocket, &(buf[sent]), &len);
      CheckAPRError(status);
      if (status != 0) break;
      sent += len;
   }
   return sent;
}

/** Send buffer wrapper that returns a future to the result of the I/O
 * operation contained in a new thread.
 * 
 * @param buf Buffer of data.
 * 
 * @param buflen Length of the data in the buffer.
 * 
 * @return A future to the number of bytes sent. */
APRFuture<apr_off_t> *TCPClient::sendBufferFuture(const char *buf, apr_off_t buflen)
{
   APRFuture<apr_off_t> *send_buffer_future = new APRFuture<apr_off_t>(mp);
   send_buffer_future->addArgument((void *)this);
   send_buffer_future->addArgument(buf, buflen);
   send_buffer_future->addArgument(buflen);
   send_buffer_future->addFunctor(getSendBufferThread());
   send_buffer_future->start();
   return send_buffer_future;
}

/** Send buffer in raw thread format
 * 
 * @param tthread The APR thread used.
 * 
 * @param args The void pointer based thread arguments.
 * 
 * @return A void pointer of 'data'. */
void *TCPClient::sendBufferThread(apr_thread_t *tthread, void *args)
{
   TCPClient *kthis = (TCPClient *)((void **)args)[0];
   const char *buf = (const char *)((void **)args)[1];
   apr_off_t buflen = *((apr_off_t *)(((void **)args)[2]));
   apr_off_t *nret = new apr_off_t();
   *nret = kthis->sendBuffer(buf, buflen);
   return nret;
}

/** Send a pipe of data over the opened connection.
 * 
 * @param aprbb An APR bucket brigade pipe of input data.
 * 
 * @return Number of bytes sent. */
apr_off_t TCPClient::sendBucketBrigade(APRBucketBrigade *aprbb)
{
   if (!isvalid) return 0;
   apr_status_t status;
   apr_size_t sent = 0;
   char *buf = (char *)malloc(sizeof(char)*MAX_BLOCK_SIZE);
   while (1)
   {
      apr_size_t len = MAX_BLOCK_SIZE;
      apr_size_t newlen = aprbb->readFromStream(buf, len);
      if ((newlen == 0) && (aprbb->isClosed())) break;
      apr_size_t tsent = 0;
      while (1)
      {
         apr_size_t newlen2 = newlen - tsent;
         if (newlen2 == 0) break;
         status = apr_socket_send(tsocket, &(buf[tsent]), &newlen2);
         CheckAPRError(status);
         if (status != 0) break;
         tsent += newlen2;
      }
      if (status != 0) break;
      aprbb->removeFromStream(newlen);
      sent += newlen;
   }
   free(buf);
   return sent;
}

/** Receive some data over the connection
 * 
 * @param len Length of data to receive.
 * 
 * @return A vector buffer of received data. */
std::vector<char> *TCPClient::receiveBuffer(apr_off_t len)
{
   if (!isvalid) return NULL;
   apr_off_t actualreq;
   std::vector<char> *returned = new std::vector<char>();
   char *buf = (char *)malloc(sizeof(char)*MAX_BLOCK_SIZE);
   while (1)
   {
      actualreq = (MAX_BLOCK_SIZE < (len - returned->size())) ? MAX_BLOCK_SIZE : (len - returned->size());
      if (len == -1) actualreq = MAX_BLOCK_SIZE;
      apr_size_t len = clientbb->popFromStream(buf, actualreq);
      if (clientbb->isClosed()) break;
      for (apr_off_t i = 0; i < len; i++)
         returned->push_back(buf[i]);
   }
   free(buf);
   return returned;
}

/** Receive buffer wrapper that returns a future to the result of the
 * I/O operation contained in a new thread.
 * 
 * @param len Length of the data to receive (and afterwards close).
 * 
 * @return A future to a buffer containing the bytes received. */
APRFuture<std::vector<char> *> *TCPClient::receiveBufferFuture(apr_off_t len)
{
   APRFuture<std::vector<char> *> *receive_buffer_future = new APRFuture<std::vector<char> *>(mp);
   receive_buffer_future->addArgument((void *)this);
   receive_buffer_future->addArgument(len);
   receive_buffer_future->addFunctor(getReceiveBufferThread());
   receive_buffer_future->start();
   return receive_buffer_future;
}

/** Receive buffer in raw thread format
 * 
 * @param tthread The APR thread used.
 * 
 * @param args The void pointer based thread arguments.
 * 
 * @return A void pointer of 'data'. */
void *TCPClient::receiveBufferThread(apr_thread_t *tthread, void *args)
{
   TCPClient *kthis = (TCPClient *)((void **)args)[0];
   apr_off_t len = *((apr_off_t *)(((void **)args)[1]));
   std::vector<char> **nret = new (std::vector<char> *)();
   *nret = kthis->receiveBuffer(len);
   return nret;
}

/** Send and then receive a buffer of data.
 * 
 * @param buf Buffer of data to be sent.
 * 
 * @param sendlen Length of data to send.
 * 
 * @param recvlen Length of data to receive.
 * 
 * @return A vector buffer of receive data. */
std::vector<char> *TCPClient::sendreceiveBuffer(const char *buf, apr_off_t sendlen, apr_off_t recvlen)
{
   if (!isvalid) return NULL;
   apr_off_t newlen = (sendlen == -1) ? strlen(buf) : sendlen;
   apr_off_t len2 = sendBuffer(buf, newlen);
   if (newlen != len2)
   {
      errprint("Could not complete data sending");
      return NULL;
   }
   std::vector<char> *returned = receiveBuffer(recvlen);
   return returned;
}

/** Send/receive buffer wrapper that returns a future to the result of
 * the I/O operation contained in a new thread.
 * 
 * @param buf Buffer of data to be sent.
 * 
 * @param sendlen Length of data to send.
 * 
 * @param recvlen Length of data to receive.
 * 
 * @return A future to a buffer containing the bytes received. */
APRFuture<std::vector<char> *> *TCPClient::sendreceiveBufferFuture(const char *buf, apr_off_t sendlen, apr_off_t recvlen)
{
   APRFuture<std::vector<char> *> *sendreceive_buffer_future = new APRFuture<std::vector<char> *>(mp);
   sendreceive_buffer_future->addArgument((void *)this);
   sendreceive_buffer_future->addArgument(buf, sendlen);
   sendreceive_buffer_future->addArgument(sendlen);
   sendreceive_buffer_future->addArgument(recvlen);
   sendreceive_buffer_future->addFunctor(getSendReceiveBufferThread());
   sendreceive_buffer_future->start();
   return sendreceive_buffer_future;
}

/** Send/receive buffer function in raw thread format
 * 
 * @param tthread The APR thread used.
 * 
 * @param args The void pointer based thread arguments.
 * 
 * @return A void pointer of 'data'. */
void *TCPClient::sendreceiveBufferThread(apr_thread_t *tthread, void *args)
{
   TCPClient *kthis = (TCPClient *)((void **)args)[0];
   const char *buf = (const char *)((void **)args)[1];
   apr_off_t sendlen = *((apr_off_t *)(((void **)args)[2]));
   apr_off_t recvlen = *((apr_off_t *)(((void **)args)[3]));
   std::vector<char> **nret = new (std::vector<char> *)();
   *nret = kthis->sendreceiveBuffer(buf, sendlen, recvlen);
   return nret;
}
